package com.company.cloud.common.mq.redis.client.consumer.thread;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.util.CollectionUtils;

import com.company.cloud.common.mq.redis.client.consumer.ISaasMqConsumer;
import com.company.cloud.common.mq.redis.client.consumer.MQConsumeMessage;
import com.company.cloud.common.mq.redis.client.consumer.MQConsumerModel;
import com.company.cloud.common.mq.redis.client.consumer.MQSaasResult;
import com.company.cloud.common.mq.redis.client.consumer.annotation.MQRedisConsumer;
import com.company.cloud.common.mq.redis.client.factory.RedisMQClientFactory;
import com.company.cloud.common.mq.redis.client.util.ThrowableUtil;

public class ConsumerRedisThread extends Thread {

	private final static Logger logger = LoggerFactory.getLogger(ConsumerRedisThread.class);

	private ISaasMqConsumer consumerHandler;

	private MQRedisConsumer mqConsumer;
	
	private MQConsumerModel consumerModel;

	private StreamOperations<String, String, String> streamOperations;

	public ConsumerRedisThread(ISaasMqConsumer consumerHandler, StreamOperations<String, String, String> streamOperations) {
		this.consumerHandler = consumerHandler;
		this.streamOperations = streamOperations;
		this.mqConsumer = consumerHandler.getClass().getAnnotation(MQRedisConsumer.class);
		this.consumerModel=new MQConsumerModel(this.mqConsumer);
		logger.debug("+++++MQ Consumer config={}",this.consumerModel.toString());
	}

	public MQRedisConsumer getMqConsumer() {
		return mqConsumer;
	}

	private List<MQConsumeMessage> readRedisMqData() {
		Consumer consumer = Consumer.from(consumerModel.getGroup(), consumerModel.getConsumer());
		StreamReadOptions readOptions = StreamReadOptions.empty();
		readOptions = readOptions.count(consumerModel.getCount());
		if (consumerModel.getBlockTimeout() > 0) {
			readOptions = readOptions.block(Duration.ofSeconds(consumerModel.getBlockTimeout()));
		}
		if (consumerModel.isAuctAck()) {
			readOptions = readOptions.autoAcknowledge();
		}
		StreamOffset<String> streams = StreamOffset.create(consumerModel.getTopic(), ReadOffset.lastConsumed());
		List<MapRecord<String, String, String>> redisMQData=null;
		try {
//			logger.debug("+++++ MQRedis Read 【读配置】,config={}",consumerModel.toString());
			redisMQData = streamOperations.read(consumer, readOptions, streams);
		} catch (Exception e) {
			// 判断是否要创建组
			if (ThrowableUtil.isNoGroupError(e)) {
				logger.debug("+++++ MQRedis Read 【读错误:-> 创建分组】,error={},config={}",e.getMessage(),consumerModel.toString());
				try {
//					String cgr =streamOperations.createGroup(this.mqConsumer.topic(), ReadOffset.latest(), this.mqConsumer.group());
					String cgr = streamOperations.createGroup(consumerModel.getTopic(), consumerModel.getGroup());
					logger.debug("++++++++ MQRedis createGroup 【创建分组配置】,result={},config={}", cgr,consumerModel.toString());
				} catch (Exception e2) {
					logger.debug("+++++ MQRedis createGroup 【创建分组错误】, error={},config={}", e2.getMessage(),consumerModel.toString());
				}
			} else {
				logger.debug("+++++ MQRedis Read 【读错误】, error:{},config={}", e.getMessage(),consumerModel.toString());
			}
		}
		List<MQConsumeMessage> data=new ArrayList<MQConsumeMessage>();
		if(!CollectionUtils.isEmpty(redisMQData)) {
//			logger.debug("read stream data : {}"+JSON.toJSONString(redisMQData));
			for(MapRecord<String, String, String> mapRecord:redisMQData) {
				MQConsumeMessage consumeMessage=new MQConsumeMessage(mapRecord.getStream(),mapRecord.getId().getValue(),mapRecord.getValue());
				data.add(consumeMessage);
			}
		}
		return data;
	}

	@Override
	public void run() {
		int waitTim = 0;
		while (!RedisMQClientFactory.clientFactoryPoolStoped) {
			List<MQConsumeMessage> mqMessageList = this.readRedisMqData();
			if(!CollectionUtils.isEmpty(mqMessageList)) {
				for (MQConsumeMessage message:mqMessageList) {
					MQSaasResult mqResult = null;
					try {
						if(consumerModel.getRunTimeout() > 0) {
							waitTim=1;
							Thread futureThread = null;
							try {
								FutureTask<MQSaasResult> futureTask = new FutureTask<MQSaasResult>(new Callable<MQSaasResult>() {
									@Override
									public MQSaasResult call() throws Exception {
										return consumerHandler.consume(message);
									}
								});
//								logger.debug("+++++ MQRedis 多线程调用 ... ");
								futureThread = new Thread(futureTask);
								futureThread.start();
								mqResult = futureTask.get(consumerModel.getRunTimeout(), TimeUnit.SECONDS);
							} catch (Exception e) {
								e.printStackTrace();
								mqResult = MQSaasResult.getFail("Timeout:" + e.getMessage());
							} finally {
								futureThread.interrupt();
							}
						} else {
							waitTim=0;
//							logger.debug("+++++ MQRedis 直接调用 ... ");
							mqResult = this.consumerHandler.consume(message);
						}
						if (mqResult == null) {
							mqResult = MQSaasResult.FAIL;
						}
					} catch (Exception e) {
						logger.error(e.getMessage(), e);
						String errorMsg = ThrowableUtil.toString(e);
						mqResult = MQSaasResult.getFail(errorMsg);
					}
					
					if(mqResult.isSuccess() && (!consumerModel.isAuctAck())) {
//						logger.debug("+++++ MQRedis acknowledge ,config={}",consumerModel.toString());
						this.streamOperations.acknowledge(consumerModel.getTopic(), consumerModel.getGroup(), mqResult.getMsgId());
					}
				}
			}else {
				 waitTim = (waitTim+5)<=60?(waitTim+5):60;
			}
			// ============================================================
			try {
				TimeUnit.SECONDS.sleep(waitTim);
			} catch (Exception e) {
				if (!RedisMQClientFactory.clientFactoryPoolStoped) {
					logger.error(e.getMessage(), e);
				}
			}
		}

	}



}
